package com.jiazhong.mingxing.boot.boot11.普通消息.消息消费;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import java.nio.charset.StandardCharsets;
import java.util.List;

@Slf4j
public class 集群消息消费1 {

    @SneakyThrows
    public static void main(String[] args) {
        // 1. 创建消费者 Consumer，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("jiazhong-consumer-a");
        // 2. 指定 NameSrv 地址
        consumer.setNamesrvAddr("124.220.170.254:9876");
        // 3. 指定消费模式：集群(CLUSTERING)和广播(BROADCASTING) 默认是集群
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 4. 订阅主题 Topic 和 Tag
        consumer.subscribe("TopicA", "sync");
        // 5. 设置回调函数，处理消息
        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
            // log.info("当有TopicA主题下的sync标签消息生产后会自动执行这里的代码");
            for (MessageExt messageExt : list) {
                String topic = messageExt.getTopic();
                String tags = messageExt.getTags();
                String body = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                log.info("主题：{},标签：{},消息：{}", topic, tags, body);
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 6. 启动消费者 Consumer
        consumer.start();
    }
}
